package com.wudl.core;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @ClassName : WordCount
 * @Description : 流式处理的一个单词统计
 * @Author :wudl
 * @Date: 2020-10-21 01:04
 */
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class WordCount {

    public static void main(String[] args) throws Exception {

        // 创建一个执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 获取数据源
        DataStreamSource<String> streamSource = env.socketTextStream("192.168.1.140", 8899);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordTuple = streamSource.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (r, out) -> {
                    String[] splitWorld = r.split(",");
                    for (String word : splitWorld) {
                        Tuple2<String, Integer> tuple = Tuple2.of(word, 1);
                        out.collect(tuple);

                    }
                }
        ).returns(new TypeHint<Tuple2<String, Integer>>() {
        });

            // 分组
        KeyedStream<Tuple2<String, Integer>, Tuple> keyedStream = wordTuple.keyBy(0);
    //  聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> resultSum = keyedStream.sum(1);
        resultSum.print();
        env.execute();



    }
}
